package com.java.code.rocketmq.trans;

import com.java.code.rocketmq.RocketmqConstants;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * [注]:
 * 1.事务消息的 Group ID 不能与其他类型消息的 Group ID 共用:与其他类型的消息不同，事务消息有回查机制，回查时服务端会根据 Group ID 去查询生产者客户端；
 * 2.特喵的,单机版还需要修改broker.conf中brokerRole=ASYNC_MASTER,否则发送消息会返回SLAVE_NOT_AVAILABLE,这样的话事务不会check
 * @author weichuanyuan
 * @e-mail wei.cy@bsoft.com.cn
 * @date 2022/09/01 20:44
 * @package com.java.code.rocketmq.trans
 * @description
 */
public class TransSend {
    public static void main(String[] args) throws Exception{
        TransactionMQProducer producer = new TransactionMQProducer(RocketmqConstants.GROUP_TRANS);
        producer.setNamesrvAddr(RocketmqConstants.NAMESVR_ADDR);
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });

        producer.setExecutorService(executorService);
        producer.setTransactionListener(new TransactionListenerImpl());
        producer.start();
//        String[] tags = new String[] {
//                RocketmqConstants.TAG_A,
//                RocketmqConstants.TAG_B,
//                RocketmqConstants.TAG_C,
//                RocketmqConstants.TAG_D,
//                RocketmqConstants.TAG_E
//        };
        for (int i = 0; i < 5; i++) {
            try {
                Message msg = new Message(RocketmqConstants.TOPIC_TRANS, RocketmqConstants.TAG_A,
                                ("Transaction RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);
                Thread.sleep(10);
            } catch (MQClientException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        producer.shutdown();
    }
}

/**
 * 事务的三个状态
 * TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
 * TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
 * TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。
 *
 * 理解的流程是这样的
 * 1.本地程序producer生产者向mq发送half消息,发送成功后,执行executeLocalTransaction开始本地事务
 * 2.返回执行成功的状态 应该是commit 或者 rollback 或者 unknow
 * 3.如果mq没有收到消息,会调用producer生产者checkLocalTransaction检查状态,本地程序producer生产者去检查本地事务的状态,给与反馈rollback或者commit或者unknow
 * 4.如果是unknow的话,重复,默认的检查次数是15次,可以修改Broker 配置文件的 transactionCheckMax修改次数
 * 5.如果是commit的话,提交事务,交给消费者
 * 6.如果是rollback的话,回滚事务,消息将不会投递给消费者
 *
 * 当发送半消息成功时，我们使用 executeLocalTransaction 方法来执行本地事务。
 * checkLocalTransaction 方法用于检查本地事务状态，并回应消息队列的检查请求
 */


class TransactionListenerImpl implements TransactionListener {
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        localTrans.put(msg.getTransactionId(), status);
        return LocalTransactionState.UNKNOW;
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    return LocalTransactionState.UNKNOW;
                case 1:
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}